package eiisan.forkjointest.countedcompleter;

import java.util.concurrent.CountedCompleter;

public class MapReducer2<E> extends CountedCompleter<E> { // version 2
    final E[] array;
    final MyMapper<E> mapper;
    final MyReducer<E> reducer;
    final int lo, hi;
    MapReducer2<E> forks, next; // record subtask forks in list
    E result;

    public MapReducer2(CountedCompleter<?> p, E[] array, MyMapper<E> mapper,
                       MyReducer<E> reducer, int lo, int hi, MapReducer2<E> next) {
        super(p);
        this.array = array;
        this.mapper = mapper;
        this.reducer = reducer;
        this.lo = lo;
        this.hi = hi;
        this.next = next;
    }

    public void compute() {
        int l = lo, h = hi;
        while (h - l >= 2) {
            int mid = (l + h) >>> 1;
            addToPendingCount(1);
            (forks = new MapReducer2(this, array, mapper, reducer, mid, h, forks)).fork();
            h = mid;
        }
        if (h > l)
            result = mapper.apply(array[l]);
        // process completions by reducing along and advancing subtask links
        for (CountedCompleter<?> c = firstComplete(); c != null; c = c.nextComplete()) {
            for (MapReducer2<E> t = (MapReducer2<E>) c, s = t.forks; s != null; s = t.forks = s.next)
                t.result = reducer.apply(t.result, s.result);
        }
    }

    public E getRawResult() {
        return result;
    }

    public static <E> E mapReduce(E[] array, MyMapper<E> mapper, MyReducer<E> reducer) {
        return new MapReducer2<E>(null, array, mapper, reducer,
                0, array.length, null).invoke();
    }
}
